<!DOCTYPE html>
<html>
<head>
  <meta charset="utf-8">
  

  
  <title>es+canal+logstash优化用户查询 | shouzimu</title>
  <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
  <meta name="description" content="记录一次使用es优化用户模糊查询的过程 相关中间件及版本如下    应用 版本    ElasticSearch 5.5   mysql 5.6   logstash -   canal 1.1.4   Springboot 2.0.5.RELEASE   替换前：mysql模糊查询用户，耗时3s+ 替换后：es查询id，再in查询数据库 100ms 思路：1用户表的更新走binlog同步，中间间">
<meta property="og:type" content="article">
<meta property="og:title" content="es+canal+logstash优化用户查询">
<meta property="og:url" content="https://shouzimu.gitee.io/2020/11/12/essync/index.html">
<meta property="og:site_name" content="shouzimu">
<meta property="og:description" content="记录一次使用es优化用户模糊查询的过程 相关中间件及版本如下    应用 版本    ElasticSearch 5.5   mysql 5.6   logstash -   canal 1.1.4   Springboot 2.0.5.RELEASE   替换前：mysql模糊查询用户，耗时3s+ 替换后：es查询id，再in查询数据库 100ms 思路：1用户表的更新走binlog同步，中间间">
<meta property="og:locale" content="zh_CN">
<meta property="article:published_time" content="2020-11-12T12:07:45.000Z">
<meta property="article:modified_time" content="2020-11-12T12:12:04.924Z">
<meta property="article:author" content="shouzimu">
<meta property="article:tag" content="elasticsearch">
<meta name="twitter:card" content="summary">
  
    <link rel="alternate" href="/atom.xml" title="shouzimu" type="application/atom+xml">
  
  
    <link rel="icon" href="/favicon.png">
  
  
    <link href="//fonts.googleapis.com/css?family=Source+Code+Pro" rel="stylesheet" type="text/css">
  
  
<link rel="stylesheet" href="/css/style.css">

<meta name="generator" content="Hexo 5.2.0"></head>

<body>
  <div id="container">
    <div id="wrap">
      <header id="header">
  <div id="banner"></div>
  <div id="header-outer" class="outer">
    <div id="header-title" class="inner">
      <h1 id="logo-wrap">
        <a href="/" id="logo">shouzimu</a>
      </h1>
      
    </div>
    <div id="header-inner" class="inner">
      <nav id="main-nav">
        <a id="main-nav-toggle" class="nav-icon"></a>
        
          <a class="main-nav-link" href="/">Home</a>
        
          <a class="main-nav-link" href="/archives">Archives</a>
        
      </nav>
      <nav id="sub-nav">
        
          <a id="nav-rss-link" class="nav-icon" href="/atom.xml" title="RSS Feed"></a>
        
        <a id="nav-search-btn" class="nav-icon" title="Search"></a>
      </nav>
      <div id="search-form-wrap">
        <form action="//google.com/search" method="get" accept-charset="UTF-8" class="search-form"><input type="search" name="q" class="search-form-input" placeholder="Search"><button type="submit" class="search-form-submit">&#xF002;</button><input type="hidden" name="sitesearch" value="https://shouzimu.gitee.io"></form>
      </div>
    </div>
  </div>
</header>
      <div class="outer">
        <section id="main"><article id="post-essync" class="article article-type-post" itemscope itemprop="blogPost">
  <div class="article-meta">
    <a href="/2020/11/12/essync/" class="article-date">
  <time datetime="2020-11-12T12:07:45.000Z" itemprop="datePublished">2020-11-12</time>
</a>
    
  </div>
  <div class="article-inner">
    
    
      <header class="article-header">
        
  
    <h1 class="article-title" itemprop="name">
      es+canal+logstash优化用户查询
    </h1>
  

      </header>
    
    <div class="article-entry" itemprop="articleBody">
      
        <p>记录一次使用es优化用户模糊查询的过程</p>
<p>相关中间件及版本如下</p>
<table>
<thead>
<tr>
<th>应用</th>
<th align="left">版本</th>
</tr>
</thead>
<tbody><tr>
<td>ElasticSearch</td>
<td align="left">5.5</td>
</tr>
<tr>
<td>mysql</td>
<td align="left">5.6</td>
</tr>
<tr>
<td>logstash</td>
<td align="left">-</td>
</tr>
<tr>
<td>canal</td>
<td align="left">1.1.4</td>
</tr>
<tr>
<td>Springboot</td>
<td align="left">2.0.5.RELEASE</td>
</tr>
</tbody></table>
<p>替换前：mysql模糊查询用户，耗时3s+</p>
<p>替换后：es查询id，再in查询数据库 100ms</p>
<h4 id="思路："><a href="#思路：" class="headerlink" title="思路："></a>思路：</h4><figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">用户表的更新走binlog同步，中间间使用canal和otter，二者介绍这里不再赘述</span><br></pre></td></tr></table></figure>



<h4 id="一、ElasticSearch配置"><a href="#一、ElasticSearch配置" class="headerlink" title="一、ElasticSearch配置"></a>一、ElasticSearch配置</h4><p>创建索引，5.5版本还需要设置type，比较蛋疼</p>
<figure class="highlight plain"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line">1、创建索引member_info</span><br><span class="line">PUT &#x2F;member_info</span><br><span class="line"></span><br><span class="line">2、设置type member</span><br><span class="line">PUT &#x2F;member_info&#x2F;_mapping&#x2F;member</span><br><span class="line">&#123;</span><br><span class="line">	&quot;properties&quot;: &#123;</span><br><span class="line">	&#125;</span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line">es工作到此结束</span><br></pre></td></tr></table></figure>



<h4 id="二、canal配置"><a href="#二、canal配置" class="headerlink" title="二、canal配置"></a>二、canal配置</h4><h5 id="1、canal-server端配置"><a href="#1、canal-server端配置" class="headerlink" title="1、canal server端配置"></a>1、canal server端配置</h5><figure class="highlight shell"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line">直接使用的单机版</span><br><span class="line"></span><br><span class="line">github下载的canal.deployer-1.1.4.tar.gz</span><br><span class="line"></span><br><span class="line">解压，编辑 conf/example/instance.properties 修改以下关键配置</span><br><span class="line">canal.instance.master.address=数据库ip:数据库端口</span><br><span class="line">canal.instance.dbUsername=数据库用户名</span><br><span class="line">canal.instance.dbPassword=数据库密码</span><br><span class="line">canal.instance.filter.regex=member_info (注：这个地方我只需要监听这一个表的数据变化)</span><br><span class="line"></span><br><span class="line">启动</span><br><span class="line">bin/startup.sh</span><br></pre></td></tr></table></figure>


<h5 id="2、canal-客户端-otter的配置，这个地方写的代码，直接上"><a href="#2、canal-客户端-otter的配置，这个地方写的代码，直接上" class="headerlink" title="2、canal 客户端 otter的配置，这个地方写的代码，直接上"></a>2、canal 客户端 otter的配置，这个地方写的代码，直接上</h5><p>MemberCanalClient.java</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br><span class="line">120</span><br><span class="line">121</span><br><span class="line">122</span><br><span class="line">123</span><br><span class="line">124</span><br><span class="line">125</span><br><span class="line">126</span><br><span class="line">127</span><br><span class="line">128</span><br><span class="line">129</span><br><span class="line">130</span><br><span class="line">131</span><br><span class="line">132</span><br><span class="line">133</span><br><span class="line">134</span><br><span class="line">135</span><br><span class="line">136</span><br><span class="line">137</span><br><span class="line">138</span><br><span class="line">139</span><br><span class="line">140</span><br><span class="line">141</span><br><span class="line">142</span><br><span class="line">143</span><br><span class="line">144</span><br><span class="line">145</span><br><span class="line">146</span><br><span class="line">147</span><br><span class="line">148</span><br><span class="line">149</span><br><span class="line">150</span><br><span class="line">151</span><br><span class="line">152</span><br><span class="line">153</span><br><span class="line">154</span><br><span class="line">155</span><br><span class="line">156</span><br><span class="line">157</span><br><span class="line">158</span><br><span class="line">159</span><br><span class="line">160</span><br><span class="line">161</span><br><span class="line">162</span><br><span class="line">163</span><br><span class="line">164</span><br><span class="line">165</span><br><span class="line">166</span><br><span class="line">167</span><br><span class="line">168</span><br><span class="line">169</span><br><span class="line">170</span><br><span class="line">171</span><br><span class="line">172</span><br><span class="line">173</span><br><span class="line">174</span><br><span class="line">175</span><br><span class="line">176</span><br><span class="line">177</span><br><span class="line">178</span><br><span class="line">179</span><br><span class="line">180</span><br><span class="line">181</span><br><span class="line">182</span><br><span class="line">183</span><br><span class="line">184</span><br><span class="line">185</span><br><span class="line">186</span><br><span class="line">187</span><br><span class="line">188</span><br><span class="line">189</span><br><span class="line">190</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="meta">@Slf4j</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">MemberCanalClient</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> TransportClient transportClient;</span><br><span class="line">    </span><br><span class="line">    String index = <span class="string">&quot;member_info&quot;</span>;</span><br><span class="line">    String type = <span class="string">&quot;member&quot;</span>;</span><br><span class="line">    </span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">initElasticSearch</span><span class="params">()</span> <span class="keyword">throws</span> UnknownHostException </span>&#123;</span><br><span class="line">        Settings settings = Settings.builder().put(<span class="string">&quot;cluster.name&quot;</span>, <span class="string">&quot;my-application-1&quot;</span>).put(<span class="string">&quot;xpack.security.user&quot;</span>, <span class="string">&quot;用户名:密码&quot;</span>).put(<span class="string">&quot;client.transport.sniff&quot;</span>, <span class="keyword">false</span>).build();</span><br><span class="line">        transportClient = <span class="keyword">new</span> PreBuiltXPackTransportClient(settings);</span><br><span class="line">        TransportAddress transportAddress = <span class="keyword">new</span> InetSocketTransportAddress(InetAddress.getByName(<span class="string">&quot;ip&quot;</span>), <span class="string">&quot;端口&quot;</span>);</span><br><span class="line">        transportClient.addTransportAddress(transportAddress);</span><br><span class="line">    &#125;</span><br><span class="line">    </span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">otterSync</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="comment">// 创建链接</span></span><br><span class="line">        CanalConnector connector = CanalConnectors.newSingleConnector(<span class="keyword">new</span> InetSocketAddress(AddressUtils.getHostIp(),</span><br><span class="line">                <span class="number">11113</span>), <span class="string">&quot;example&quot;</span>, <span class="string">&quot;&quot;</span>, <span class="string">&quot;&quot;</span>);</span><br><span class="line">        <span class="keyword">int</span> batchSize = <span class="number">1000</span>;</span><br><span class="line">        <span class="keyword">int</span> emptyCount = <span class="number">0</span>;</span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            connector.connect();</span><br><span class="line">            connector.subscribe(<span class="string">&quot;member.member_info&quot;</span>);</span><br><span class="line">            connector.rollback();</span><br><span class="line">            <span class="keyword">while</span> (<span class="keyword">true</span>) &#123;</span><br><span class="line">                <span class="comment">// 获取指定数量的数据</span></span><br><span class="line">                Message message = connector.getWithoutAck(batchSize);</span><br><span class="line">                <span class="keyword">long</span> batchId = message.getId();</span><br><span class="line">                <span class="keyword">int</span> size = message.getEntries().size();</span><br><span class="line">                <span class="keyword">if</span> (batchId == -<span class="number">1</span> || size == <span class="number">0</span>) &#123;</span><br><span class="line">                    emptyCount++;</span><br><span class="line">                    System.out.println(<span class="string">&quot;empty count : &quot;</span> + emptyCount);</span><br><span class="line">                    <span class="keyword">try</span> &#123;</span><br><span class="line">                        Thread.sleep(<span class="number">1000</span>);</span><br><span class="line">                    &#125; <span class="keyword">catch</span> (InterruptedException e) &#123;</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                    emptyCount = <span class="number">0</span>;</span><br><span class="line">                    <span class="comment">// System.out.printf(&quot;message[batchId=%s,size=%s] \n&quot;, batchId, size);</span></span><br><span class="line">                    printEntry(message.getEntries());</span><br><span class="line">                &#125;</span><br><span class="line">    </span><br><span class="line">                connector.ack(batchId); <span class="comment">// 提交确认</span></span><br><span class="line">                <span class="comment">// connector.rollback(batchId); // 处理失败, 回滚数据</span></span><br><span class="line">            &#125;</span><br><span class="line">        &#125; <span class="keyword">finally</span> &#123;</span><br><span class="line">            connector.disconnect();</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">    </span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">printEntry</span><span class="params">(List&lt;Entry&gt; entrys)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">for</span> (Entry entry : entrys) &#123;</span><br><span class="line">            <span class="keyword">if</span> (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) &#123;</span><br><span class="line">                <span class="keyword">continue</span>;</span><br><span class="line">            &#125;</span><br><span class="line">    </span><br><span class="line">            RowChange rowChage = <span class="keyword">null</span>;</span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">                rowChage = RowChange.parseFrom(entry.getStoreValue());</span><br><span class="line">            &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">                <span class="keyword">throw</span> <span class="keyword">new</span> RuntimeException(<span class="string">&quot;ERROR ## parser of eromanga-event has an error , data:&quot;</span> + entry.toString(),</span><br><span class="line">                        e);</span><br><span class="line">            &#125;</span><br><span class="line">    </span><br><span class="line">            EventType eventType = rowChage.getEventType();</span><br><span class="line">            System.out.println(String.format(<span class="string">&quot;================&amp;gt; binlog[%s:%s] , name[%s,%s] , eventType : %s&quot;</span>,</span><br><span class="line">                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),</span><br><span class="line">                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),</span><br><span class="line">                    eventType));</span><br><span class="line">    </span><br><span class="line">            <span class="keyword">for</span> (RowData rowData : rowChage.getRowDatasList()) &#123;</span><br><span class="line">                <span class="keyword">try</span> &#123;</span><br><span class="line">                    operator(rowData, eventType);</span><br><span class="line">                &#125;<span class="keyword">catch</span> (Exception e)&#123;</span><br><span class="line">    </span><br><span class="line">                &#125;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">    </span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">printColumn</span><span class="params">(List&lt;Column&gt; columns)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">for</span> (Column column : columns) &#123;</span><br><span class="line">            System.out.println(column.getName() + <span class="string">&quot; : &quot;</span> + column.getValue() + <span class="string">&quot;    update=&quot;</span> + column.getUpdated());</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">operator</span><span class="params">(RowData rowData, EventType eventType)</span> </span>&#123;</span><br><span class="line">        System.out.println(eventType);</span><br><span class="line">        List&lt;Column&gt; columnList = rowData.getAfterColumnsList();</span><br><span class="line">        <span class="keyword">if</span> (eventType == EventType.DELETE) &#123;</span><br><span class="line">            columnList = rowData.getBeforeColumnsList();</span><br><span class="line">        &#125;</span><br><span class="line">        Member m = <span class="keyword">new</span> Member();</span><br><span class="line">        <span class="keyword">boolean</span> update = <span class="keyword">false</span>;</span><br><span class="line">        <span class="keyword">for</span> (Column column : columnList) &#123;</span><br><span class="line">            System.out.println(column.getName() + <span class="string">&quot; : &quot;</span> + column.getValue() + <span class="string">&quot;    update=&quot;</span> + column.getUpdated());</span><br><span class="line">    </span><br><span class="line">            <span class="keyword">switch</span> (column.getName()) &#123;</span><br><span class="line">                <span class="keyword">case</span> <span class="string">&quot;id&quot;</span>:</span><br><span class="line">                    m.setId(column.getValue());</span><br><span class="line">                    update = checkUpdate(update, column.getUpdated());</span><br><span class="line">                    <span class="keyword">break</span>;</span><br><span class="line">                <span class="keyword">case</span> <span class="string">&quot;mobile&quot;</span>:</span><br><span class="line">                    m.setMobile(column.getValue());</span><br><span class="line">                    update = checkUpdate(update, column.getUpdated());</span><br><span class="line">                    <span class="keyword">break</span>;</span><br><span class="line">                <span class="keyword">case</span> <span class="string">&quot;name&quot;</span>:</span><br><span class="line">                    m.setName(column.getValue());</span><br><span class="line">                    update = checkUpdate(update, column.getUpdated());</span><br><span class="line">                    <span class="keyword">break</span>;</span><br><span class="line">                <span class="keyword">case</span> <span class="string">&quot;pinyin_code&quot;</span>:</span><br><span class="line">                    m.setPy(column.getValue());</span><br><span class="line">                    update = checkUpdate(update, column.getUpdated());</span><br><span class="line">                    <span class="keyword">break</span>;</span><br><span class="line">                <span class="keyword">case</span> <span class="string">&quot;is_deleted&quot;</span>:</span><br><span class="line">                    m.setIsDeleted(column.getValue());</span><br><span class="line">                    update = checkUpdate(update, column.getUpdated());</span><br><span class="line">                    <span class="keyword">break</span>;</span><br><span class="line">                <span class="keyword">default</span>:</span><br><span class="line">                    <span class="keyword">break</span>;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="comment">//关键字段更新才执行后续操作</span></span><br><span class="line">        <span class="keyword">if</span> (update) &#123;</span><br><span class="line">            log.info(<span class="string">&quot;sync member:&#123;&#125;&quot;</span>, m);</span><br><span class="line">            <span class="keyword">if</span> (eventType == EventType.DELETE || Objects.equals(m.getIsDeleted(), <span class="string">&quot;1&quot;</span>)) &#123;</span><br><span class="line">                deleteById(m.getId());</span><br><span class="line">            &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                saveOrUpdateIndex(m);</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    </span><br><span class="line">    &#125;</span><br><span class="line">    </span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">boolean</span> <span class="title">checkUpdate</span><span class="params">(<span class="keyword">boolean</span> u, <span class="keyword">boolean</span> updated)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">if</span> (u || updated) &#123;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">true</span>;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">    &#125;</span><br><span class="line">    </span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">saveOrUpdateIndex</span><span class="params">(Member member)</span> </span>&#123;</span><br><span class="line">        XContentBuilder builder;</span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            builder = XContentFactory.jsonBuilder()</span><br><span class="line">                    .startObject()</span><br><span class="line">                    .field(<span class="string">&quot;id&quot;</span>, member.getId())</span><br><span class="line">                    .field(<span class="string">&quot;mobile&quot;</span>, member.getMobile())</span><br><span class="line">                    .field(<span class="string">&quot;name&quot;</span>, member.getName())</span><br><span class="line">                    .field(<span class="string">&quot;py&quot;</span>, member.getPy())</span><br><span class="line">                    .endObject();</span><br><span class="line">        &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">            e.printStackTrace();</span><br><span class="line">            <span class="keyword">return</span>;</span><br><span class="line">        &#125;</span><br><span class="line">    </span><br><span class="line">        GetResponse getResponse = searchById(member.getId());</span><br><span class="line">        <span class="keyword">if</span> (<span class="keyword">null</span> == getResponse || <span class="keyword">null</span> == getResponse.getSourceAsString()) &#123;</span><br><span class="line">            <span class="comment">//创建索引</span></span><br><span class="line">            IndexResponse indexResponse = transportClient.prepareIndex(index, type, member.getId()).setSource(builder).execute().actionGet();</span><br><span class="line">            System.out.println(indexResponse);</span><br><span class="line">    </span><br><span class="line">        &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">            <span class="comment">//更新索引</span></span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">                UpdateResponse response = transportClient.prepareUpdate(index, type, member.getId()).setDoc(builder).get();</span><br><span class="line">                System.out.println(response);</span><br><span class="line">            &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">                e.printStackTrace();</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line"></span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> GetResponse <span class="title">searchById</span><span class="params">(String id)</span> </span>&#123;</span><br><span class="line">        GetResponse response = transportClient.prepareGet(index, type, id).execute().actionGet();</span><br><span class="line">        <span class="keyword">return</span> response;</span><br><span class="line">    &#125;</span><br><span class="line">    </span><br><span class="line">    <span class="function"><span class="keyword">public</span> DeleteResponse <span class="title">deleteById</span><span class="params">(String id)</span> </span>&#123;</span><br><span class="line">        DeleteResponse response = transportClient.prepareDelete(index, type, id).execute().actionGet();</span><br><span class="line">        log.info(<span class="string">&quot;deleteById:&#123;&#125;,response:&#123;&#125;&quot;</span>, id, response);</span><br><span class="line">        <span class="keyword">return</span> response;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>Member.java</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Member</span> <span class="keyword">implements</span> <span class="title">Serializable</span> </span>&#123;</span><br><span class="line">    <span class="keyword">private</span> String id;</span><br><span class="line">    <span class="keyword">private</span> String mobile;</span><br><span class="line">    <span class="keyword">private</span> String name;</span><br><span class="line">    <span class="keyword">private</span> String py;</span><br><span class="line">    <span class="keyword">private</span> String isDeleted;</span><br><span class="line"> &#125;</span><br></pre></td></tr></table></figure>

<p>canal client使用springboot启动，此处不展开</p>
<p>启动clinet</p>
<h4 id="三、logstash的处理"><a href="#三、logstash的处理" class="headerlink" title="三、logstash的处理"></a>三、logstash的处理</h4><p>canal只能处理增量的数据，那么历史的怎么办呢，这个地方使用logstash导入，配置文件如下</p>
<p>sync.conf</p>
<figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br></pre></td><td class="code"><pre><span class="line"><span class="attr">input</span> <span class="string">&#123;</span></span><br><span class="line">    <span class="attr">stdin</span> <span class="string">&#123;</span></span><br><span class="line">    <span class="attr">&#125;</span></span><br><span class="line">    <span class="attr">jdbc</span> <span class="string">&#123;</span></span><br><span class="line"><span class="comment">      # mysql 数据库链接</span></span><br><span class="line">      <span class="attr">jdbc_connection_string</span> =<span class="string">&gt; &quot;jdbc:mysql://数据库连接&quot;</span></span><br><span class="line"><span class="comment">      # 用户名和密码</span></span><br><span class="line">      <span class="attr">jdbc_user</span> =<span class="string">&gt; &quot;数据库用户名&quot;</span></span><br><span class="line">      <span class="attr">jdbc_password</span> =<span class="string">&gt; &quot;数据库密码&quot;</span></span><br><span class="line"><span class="comment">      # 驱动</span></span><br><span class="line">      <span class="attr">jdbc_driver_library</span> =<span class="string">&gt; &quot;/usr/local/logstash/mysqldoc/mysql-connector-java-8.0.8-dmr.jar&quot;</span></span><br><span class="line"><span class="comment">      # 驱动类名</span></span><br><span class="line">      <span class="attr">jdbc_driver_class</span> =<span class="string">&gt; &quot;com.mysql.jdbc.Driver&quot;</span></span><br><span class="line">      <span class="attr">jdbc_paging_enabled</span> =<span class="string">&gt; &quot;true&quot;</span></span><br><span class="line">      <span class="attr">jdbc_page_size</span> =<span class="string">&gt; &quot;50000&quot;</span></span><br><span class="line"><span class="comment">      # 执行的sql</span></span><br><span class="line">      <span class="attr">statement</span> =<span class="string">&gt; &quot;SELECT m.id as id,m.pinyin_code as py,m.name,m.mobile FROM 用户表 m WHERE length(m.name) &gt; 1 and is_deleted=0   ORDER BY m.create_time DESC&quot;</span></span><br><span class="line">      <span class="attr">type</span> =<span class="string">&gt; &quot;es创建的type&quot;</span></span><br><span class="line"> <span class="attr">&#125;</span></span><br><span class="line"><span class="attr">&#125;</span></span><br><span class="line"></span><br><span class="line"><span class="attr">filter</span> <span class="string">&#123;</span></span><br><span class="line">    <span class="attr">json</span> <span class="string">&#123;</span></span><br><span class="line">        <span class="attr">source</span> =<span class="string">&gt; &quot;message&quot;</span></span><br><span class="line">        <span class="attr">remove_field</span> =<span class="string">&gt; [&quot;message&quot;]</span></span><br><span class="line">    <span class="attr">&#125;</span></span><br><span class="line"><span class="attr">&#125;</span></span><br><span class="line"></span><br><span class="line"><span class="attr">output</span> <span class="string">&#123;</span></span><br><span class="line">    <span class="attr">elasticsearch</span> <span class="string">&#123;</span></span><br><span class="line">        <span class="attr">hosts</span> =<span class="string">&gt; [&quot;es服务器&quot;]</span></span><br><span class="line">        <span class="attr">index</span> =<span class="string">&gt; &quot;创建的所以&quot;</span></span><br><span class="line">        <span class="attr">user</span>  =<span class="string">&gt; &quot;xpack用户名&quot;</span></span><br><span class="line">        <span class="attr">password</span> =<span class="string">&gt; &quot;xpack密码&quot;</span></span><br><span class="line">        <span class="attr">document_id</span> =<span class="string">&gt; &quot;%&#123;id&#125;&quot;</span></span><br><span class="line">    <span class="attr">&#125;</span></span><br><span class="line">    <span class="attr">stdout</span> <span class="string">&#123;</span></span><br><span class="line">        <span class="attr">codec</span> =<span class="string">&gt; json_lines</span></span><br><span class="line">    <span class="attr">&#125;</span></span><br><span class="line"><span class="attr">&#125;</span></span><br></pre></td></tr></table></figure>

<p>启动logstatsh同步</p>
<figure class="highlight shell"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">bin/logstash -f sync.conf</span><br></pre></td></tr></table></figure>

<h4 id="四、查询代码"><a href="#四、查询代码" class="headerlink" title="四、查询代码"></a>四、查询代码</h4><p>从es中查询数据</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Override</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> List&lt;Member&gt; <span class="title">queryMember</span><span class="params">(String keyWord)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">long</span> start = System.currentTimeMillis();</span><br><span class="line">        <span class="comment">//搜索用户</span></span><br><span class="line">        List&lt;Member&gt; elasticList = <span class="keyword">new</span> ArrayList&lt;&gt;(<span class="number">30</span>);</span><br><span class="line">        <span class="keyword">if</span>(StringUtils.isBlank(keyWord))&#123;</span><br><span class="line">            <span class="keyword">return</span> elasticList;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="comment">//ES检索并排序</span></span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line"></span><br><span class="line">            BoolQueryBuilder booleanQuery = QueryBuilders.boolQuery();</span><br><span class="line"></span><br><span class="line">            QueryBuilder match = QueryBuilders.queryStringQuery(keyWord)</span><br><span class="line">                    .field(<span class="string">&quot;name&quot;</span>).field(<span class="string">&quot;mobile&quot;</span>).field(<span class="string">&quot;py&quot;</span>).defaultOperator(Operator.AND).boost(<span class="number">100</span>);</span><br><span class="line">            booleanQuery.should(match);</span><br><span class="line"></span><br><span class="line">            QueryBuilder like =  QueryBuilders.queryStringQuery(keyWord+<span class="string">&quot;*&quot;</span>)</span><br><span class="line">                    .field(<span class="string">&quot;name&quot;</span>).field(<span class="string">&quot;mobile&quot;</span>).field(<span class="string">&quot;py&quot;</span>).defaultOperator(Operator.AND).boost(<span class="number">50</span>);</span><br><span class="line">            booleanQuery.should(like);</span><br><span class="line"></span><br><span class="line"></span><br><span class="line">            QueryBuilder likeAll =  QueryBuilders.queryStringQuery(<span class="string">&quot;*&quot;</span>+keyWord+<span class="string">&quot;*&quot;</span>)</span><br><span class="line">                    .field(<span class="string">&quot;name&quot;</span>).field(<span class="string">&quot;mobile&quot;</span>).field(<span class="string">&quot;py&quot;</span>).defaultOperator(Operator.AND).boost(<span class="number">50</span>);</span><br><span class="line">            booleanQuery.should(likeAll);</span><br><span class="line"></span><br><span class="line"></span><br><span class="line"></span><br><span class="line">            SearchHits searchHits = transportClient.prepareSearch(memberIndex)</span><br><span class="line">                    .setSearchType(SearchType.QUERY_THEN_FETCH)</span><br><span class="line">                    .setTimeout(TimeValue.timeValueSeconds(<span class="number">3</span>)).</span><br><span class="line">                            setExplain(<span class="keyword">true</span>).setQuery(booleanQuery).setFrom(<span class="number">0</span>).setSize(<span class="number">30</span>).get()</span><br><span class="line">                    .getHits();</span><br><span class="line">            searchHits.forEach(searchHitFields -&gt; &#123;</span><br><span class="line">                Map&lt;String, Object&gt; map = searchHitFields.getSource();</span><br><span class="line">                String name = (String) map.get(<span class="string">&quot;name&quot;</span>);</span><br><span class="line">                String mobile = (String) map.get(<span class="string">&quot;mobile&quot;</span>);</span><br><span class="line">                String py = (String) map.get(<span class="string">&quot;py&quot;</span>);</span><br><span class="line">                String id = (String) map.get(<span class="string">&quot;id&quot;</span>);</span><br><span class="line">                Member m = <span class="keyword">new</span> Member();</span><br><span class="line">                m.setId(id);</span><br><span class="line">                m.setName(name);</span><br><span class="line">                m.setMobile(mobile);</span><br><span class="line">                m.setPy(py);</span><br><span class="line">                elasticList.add(m);</span><br><span class="line">            &#125;);</span><br><span class="line">        &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">            log.error(<span class="string">&quot;es搜索异常:&#123;&#125;&quot;</span>, ExceptionStackUtil.getExceptionStackTrace(e));</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">long</span> end = System.currentTimeMillis();</span><br><span class="line">        log.info(<span class="string">&quot;从es中搜索出来的值 key:&#123;&#125; time:&#123;&#125;ms res：&#123;&#125;&quot;</span>,keyWord, end - start, elasticList.size());</span><br><span class="line">        <span class="keyword">return</span> elasticList;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

      
    </div>
    <footer class="article-footer">
      <a data-url="https://shouzimu.gitee.io/2020/11/12/essync/" data-id="ckhesflrz0007zoiz1comh6sf" class="article-share-link">Share</a>
      
      
  <ul class="article-tag-list" itemprop="keywords"><li class="article-tag-list-item"><a class="article-tag-list-link" href="/tags/elasticsearch/" rel="tag">elasticsearch</a></li></ul>

    </footer>
  </div>
  
    
<nav id="article-nav">
  
  
    <a href="/2020/11/12/centos-static-ip/" id="article-nav-older" class="article-nav-link-wrap">
      <strong class="article-nav-caption">Older</strong>
      <div class="article-nav-title">centos配置静态ip</div>
    </a>
  
</nav>

  
</article>

</section>
        
          <aside id="sidebar">
  
    

  
    
  <div class="widget-wrap">
    <h3 class="widget-title">Tags</h3>
    <div class="widget">
      <ul class="tag-list" itemprop="keywords"><li class="tag-list-item"><a class="tag-list-link" href="/tags/cloud/" rel="tag">cloud</a></li><li class="tag-list-item"><a class="tag-list-link" href="/tags/elasticsearch/" rel="tag">elasticsearch</a></li><li class="tag-list-item"><a class="tag-list-link" href="/tags/esgynDB/" rel="tag">esgynDB</a></li><li class="tag-list-item"><a class="tag-list-link" href="/tags/java/" rel="tag">java</a></li><li class="tag-list-item"><a class="tag-list-link" href="/tags/os/" rel="tag">os</a></li></ul>
    </div>
  </div>


  
    
  <div class="widget-wrap">
    <h3 class="widget-title">Tag Cloud</h3>
    <div class="widget tagcloud">
      <a href="/tags/cloud/" style="font-size: 10px;">cloud</a> <a href="/tags/elasticsearch/" style="font-size: 10px;">elasticsearch</a> <a href="/tags/esgynDB/" style="font-size: 10px;">esgynDB</a> <a href="/tags/java/" style="font-size: 10px;">java</a> <a href="/tags/os/" style="font-size: 10px;">os</a>
    </div>
  </div>

  
    
  <div class="widget-wrap">
    <h3 class="widget-title">Archives</h3>
    <div class="widget">
      <ul class="archive-list"><li class="archive-list-item"><a class="archive-list-link" href="/archives/2020/11/">十一月 2020</a></li></ul>
    </div>
  </div>


  
    
  <div class="widget-wrap">
    <h3 class="widget-title">Recent Posts</h3>
    <div class="widget">
      <ul>
        
          <li>
            <a href="/2020/11/12/essync/">es+canal+logstash优化用户查询</a>
          </li>
        
          <li>
            <a href="/2020/11/12/centos-static-ip/">centos配置静态ip</a>
          </li>
        
          <li>
            <a href="/2020/11/12/remote-debug/">idea远程调试</a>
          </li>
        
          <li>
            <a href="/2020/11/12/k8s-single/">Centos7安装单节点kubernetes</a>
          </li>
        
          <li>
            <a href="/2020/11/12/cdh-esgyn-change-ip/">CDH+esgynDB 修改ip</a>
          </li>
        
      </ul>
    </div>
  </div>

  
</aside>
        
      </div>
      <footer id="footer">
  
  <div class="outer">
    <div id="footer-info" class="inner">
      &copy; 2020 shouzimu<br>
      Powered by <a href="http://hexo.io/" target="_blank">Hexo</a>
    </div>
  </div>
</footer>
    </div>
    <nav id="mobile-nav">
  
    <a href="/" class="mobile-nav-link">Home</a>
  
    <a href="/archives" class="mobile-nav-link">Archives</a>
  
</nav>
    

<script src="//ajax.googleapis.com/ajax/libs/jquery/2.0.3/jquery.min.js"></script>


  
<link rel="stylesheet" href="/fancybox/jquery.fancybox.css">

  
<script src="/fancybox/jquery.fancybox.pack.js"></script>




<script src="/js/script.js"></script>




  </div>
</body>
</html>